package com.yeyks.commonReference.append.rocketmq.service.producer;

import cn.hutool.json.JSONUtil;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.OnExceptionContext;
import com.aliyun.openservices.ons.api.SendCallback;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.yeyks.common.constants.rocketmq.TopicAndTag;
import com.yeyks.common.dto.mq.*;
import com.yeyks.commonReference.append.rocketmq.config.ProducerRocketMqConfig;
import com.yeyks.commonReference.append.rocketmq.pojo.RocketMQWebsocketBasePojo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;

/**
 * @author fuchangshuai
 * @date 2019/08/09 14:30
 */
@Service
@Slf4j
public class MQProducerService {

    @Autowired(required = false)
    private ProducerRocketMqConfig rocketMqConfig;

    /**
     * web给c端发消息使用(只限抢单)
     */
    public void webToCSendSimpleMsg(String msg) {
        sendSimpleMsg(msg, null, TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_WEB_TO_C_TAG);
    }

    /**
     * web给B端发消息使用(cn酒店更新使用)
     */
    public void webToBCnSendMsg(List<WebToBCnMessage> msg) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), null, TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_WEB_TO_B_FOR_CN_HOTEL_SYNC);
    }

    /**
     * 消息推送（例如活动推送）
     */
    public void webToCSendMsg(WebToCSendMessage msg) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), null, TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_WEB_TO_C_SEND_TAG);
    }

    /**
     * c端发送抢单消息
     */
    public void sendGrabTheOrderCToBMsg(GrabTheOrderCToBMessage msg) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), null, TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_GRAB_THE_ORDER);
    }

    /**
     * c端发送抢单成功消息
     */
    public void sendGrabTheOrderSuccessMsg(GrabTheOrderSuccessMessage msg) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), null, TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_GRAB_THE_ORDER_SUCCESS);
    }

    /**
     * c端内部websocket群发消息
     */
    public void sendGrabTheOrderWebsocketMsg(RocketMQWebsocketBasePojo msg) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), null, TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_GRAB_THE_ORDER_WEBSOCKET);
    }

    /**
     * 服务器发送websocket群发消息
     */
    public void sendWebsocketMsg(RocketMQWebsocketBasePojo msg) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), null, TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_WEBSOCKET);
    }

    /**
     * 退款c to web通知
     */
    public void sendRefundNoticeMsg(RefundNoticeMessage msg) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), null, TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_REFUND_NOTICE);
    }

    /**
     * 抢单发mq到b端
     */
    public void sendRobToBMsg(RocketMQWebsocketBasePojo msg) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), null, TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_ROB_TO_B_TAG);
    }

    /**
     * 抢单结果发mq到b端
     */
    public void sendRobResultMsg(RocketMQWebsocketBasePojo msg) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), null, TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_ROB_RESULT_TO_B_TAG);
    }


    /**
     * c端发送抢单超时检查消息
     */
    public void sendGrabTheOrderCheckMsg(GrabTheOrderCheckMessage msg) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), 60 * 10, TopicAndTag.TOPIC_YYKS_DELAY, TopicAndTag.TAG_GRAB_THE_ORDER_CHECK);
    }

    /**
     * c端发送延迟订单30分钟检测消息
     */
    public void sendOrderCheckToWebMessage(OrderCheckToWebMessage msg) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), 60 * 30, TopicAndTag.TOPIC_YYKS_DELAY, TopicAndTag.TAG_CHECK_NOT_PAY);
    }

    /**
     * es更新hotel,room,calendar
     */
    public void sendEsUpdate(EsUpdateMessage msg, Integer delayTime) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), delayTime, TopicAndTag.TOPIC_YYKS_DELAY, TopicAndTag.TAG_ESUPDATE);
    }

    /**
     * 线下会员卡支付成功发佣金
     */
    public void sendMiniCardOrderSuccessMessage(MiniCardOrderSuccessMessage msg) {
        sendSimpleMsg(JSONUtil.toJsonStr(msg), null, TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_CARD_PAY_SUCCESS);
    }

//    /**
//     * 小黑鱼礼包订单检查
//     */
//    public void sendDelayExchangeRechargeBiggit(String orderNo) {
//        sendSimpleMsg(orderNo, 60 * 30, TopicAndTag.TOPIC_YYKS_DELAY, TopicAndTag.TAG_BIGGIT_ORDER_CHECK);
//    }

    /**
     * 同步发送实体对象消息
     * 可靠同步发送：同步发送是指消息发送方发出数据后，会在收到接收方发回响应之后才发下一个数据包的通讯方式；
     * 特点：速度快；有结果反馈；数据可靠；
     * 应用场景：应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等；
     */
    public void sendSimpleMsg(String msg, String topic, String tag) {
        sendSimpleMsg(msg, null, topic, tag);
    }

    /**
     * 发送简单带延迟的mq消息
     *
     * @param msg
     * @param delayTime 延迟的时间(单位/秒)
     */
    public void sendSimpleMsg(String msg, Integer delayTime, String topic, String tag) {
        Message message = new Message(topic, tag, msg.getBytes());
        if (null != delayTime) {
            // 延时时间单位为毫秒（ms），指定一个时刻，在这个时刻之后才能被消费，3表示3秒后才能被消费
            delayTime = 1000 * delayTime;
            message.setStartDeliverTime(System.currentTimeMillis() + delayTime);
        }
        try {
            SendResult sendResult = rocketMqConfig.getProducer().send(message);
            assert sendResult != null;
//            log.info(new Date() + " Send mq success! Topic is:" + topic + " msgId is: " + sendResult.getMessageId() + ",msg = " + msg);
            StringBuffer sbStr = new StringBuffer(" ");
            sbStr.append(new Date()).append("====");
            sbStr.append(" Send mq success!");
            sbStr.append(" Topic and tag is: ").append(topic).append("_").append(tag);
            sbStr.append(" message is:").append(msg);
            sbStr.append(" msgId is: ").append(sendResult.getMessageId());
            log.info(sbStr.toString());
        } catch (ONSClientException e) {
            // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
            log.info(new Date() + " Send mq failed! Topic and tag is:" + topic+"_"+tag);
            e.printStackTrace();
        }
    }

    /**
     * 异步发送消息
     * 可靠异步发送：发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式；
     * 特点：速度快；有结果反馈；数据可靠；
     * 应用场景：异步发送一般用于链路耗时较长,对 rt响应时间较为敏感的业务场景,例如用户视频上传后通知启动转码服务,转码完成后通知推送转码结果等；
     *
     * @param msg
     * @return
     */
    public boolean sendMsgAsy(String msg, String topic, String tag) {
        Long startTime = System.currentTimeMillis();
        Message message = new Message(topic, tag, msg.getBytes());
        rocketMqConfig.getProducer().sendAsync(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                ///消息发送成功
                log.info("send message success. topic=" + sendResult.getMessageId());
            }

            @Override
            public void onException(OnExceptionContext context) {
                //消息发送失败
                log.info("send message failed. execption=" + context.getException());
            }
        });
        Long endTime = System.currentTimeMillis();
        log.info("单次生产耗时：" + (endTime - startTime) / 1000);
        return true;
    }

}
